package com.yjy.java8_CompletableFuture;

import org.junit.Test;

import java.util.Random;
import java.util.concurrent.*;

public class MainTest {

    // supplyAsync()创建异步任务
    @Test
    public void test_create() {
        // supplyAsync() 创建异步任务，有返回值
        CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            boolean flag = new Random().nextBoolean();
            if (flag) {
                throw new RuntimeException("测试抛出异常");
            }
            System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
            return 123;
        });

        System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
        try {
            // 等待子任务执行完成
            System.out.println("运行结果：" + cf.get());
        } catch (Exception e) {
            System.out.println("子线程抛出异常：" + e.getMessage());
        }

        System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
    }

    @Test
    public void test_create_v2() {
        // runAsync() 创建异步任务，无返回值
        CompletableFuture cf = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            boolean flag = new Random().nextBoolean();
            if (flag) {
                throw new RuntimeException("测试抛出异常");
            }
            System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
        });

        System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
        try {
            // 等待子任务执行完成
            System.out.println("运行结果：" + cf.get());
        } catch (Exception e) {
            System.out.println("子线程抛出异常：" + e.getMessage());
        }

        System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
    }

    // 指定线程池执行
    @Test
    public void test_create_v3() {
//        ForkJoinPool pool = new ForkJoinPool();
        ExecutorService pool = Executors.newFixedThreadPool(2);
        CompletableFuture cf = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            boolean flag = new Random().nextBoolean();
            if (flag) {
                throw new RuntimeException("测试抛出异常");
            }
            System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
        }, pool);

        System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
        try {
            // 等待子任务执行完成
            System.out.println("运行结果：" + cf.get());
        } catch (Exception e) {
            System.out.println("子线程抛出异常：" + e.getMessage());
        }

        System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
    }

    // 异步回调
    // thenApply() 表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果（即方法返回值）作为入参传递到回调方法中。
    @Test
    public void test_callback_thenApply() {
        CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
            return new Random().nextInt(20);
        });

        // 将上面的返回值传入传入thenAccept()中
        CompletableFuture<String> cf2 = cf.thenApply(result -> {
            System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
            System.out.println("result=" + result);
            System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
            return "test" + result;
        });

        System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
        try {
            // 等待子任务执行完成
            System.out.println("运行结果：" + cf.get());
        } catch (Exception e) {
            System.out.println("子线程抛出异常：" + e.getMessage());
        }
        try {
            // 等待子任务执行完成
            System.out.println("运行结果：" + cf2.get());
        } catch (Exception e) {
            System.out.println("子线程抛出异常：" + e.getMessage());
        }
        System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
    }

    @Test
    public void test() {
        // supplyAsync() 创建异步任务，有返回值
        CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            boolean flag = new Random().nextBoolean();
            if (flag) {
                throw new RuntimeException("测试抛出异常");
            }
            System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
            return 123;
        });
        CompletableFuture<Integer> cf2 = cf.whenComplete((result, e) -> {
            if (e != null) {
                System.out.println("抛出异常");
                e.printStackTrace();
            } else {
                System.out.println("结果为:" + result);
            }
        });
        System.out.println(Thread.currentThread() + " 开始, time=" + System.currentTimeMillis());
        try {
            // 等待子任务执行完成
            System.out.println("运行结果：" + cf.get());
        } catch (Exception e) {
            System.out.println("子线程抛出异常：" + e.getMessage());
        }
        try {
            // 等待子任务执行完成
            System.out.println("运行结果：" + cf2.get());
        } catch (Exception e) {
            System.out.println("子线程抛出异常：" + e.getMessage());
        }
        System.out.println(Thread.currentThread() + " 退出, time=" + System.currentTimeMillis());
    }


    // 首秀
    @Test
    public void teest11() throws InterruptedException {
        // 创建异步任务
        CompletableFuture<Double> completableFuture = CompletableFuture.supplyAsync(MainTest::fetchPrice);
        // 如果执行成功
        completableFuture.thenAccept(result -> {
            System.out.println("price：" + result);
        });
        // 如果执行异常
        completableFuture.exceptionally(e -> {
            e.printStackTrace();
            return null;
        });

        // 主线程不要立即结束，否则completableFuture默认使用的线程池会立刻关闭
        Thread.sleep(1000);
    }

    static Double fetchPrice() {
        try {
            Thread.sleep(100);
        } catch (Exception e) {

        }
        if (Math.random() < 0.3) {
            throw new RuntimeException("fetch price failed!");
        }
        return 5 + Math.random() * 20;
    }

    // 多个CompletableFuture串行执行
    // 需求描述：第一个CompletableFuture根据证券名称查询证券代码，第二个CompletableFuture根据证券代码查询证券价格，两个串行化执行
    @Test
    public void test12() throws InterruptedException {
        // 第一个任务:
        CompletableFuture<String> firstCF = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油");
        });
        // cfQuery成功后继续执行下一个任务:
        CompletableFuture<Double> secondCF = firstCF.thenApplyAsync((code) -> {
            return fetchPrice(code);
        });
        // cfFetch成功后打印结果:
        secondCF.thenAccept((result) -> {
            System.out.println("price: " + result);
        });
        // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
        Thread.sleep(2000);
    }

    static String queryCode(String name) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
        }
        return "601857";
    }

    static Double fetchPrice(String code) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
        }
        return 5 + Math.random() * 20;
    }

    // 多个CompletableFuture并行执行
    // 需求：同时从新浪和网易查询证券代码，只要任意一个返回结果，就进行下一步查询价格，查询价格也同时从新浪和网易查询，只要任意一个返回结果
    @Test
    public void test13() throws InterruptedException {
        // 两个CompletableFuture异步查询
        CompletableFuture<String> sinaCF = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油", "https://finance.sina.com.cn/code/");
        });
        CompletableFuture<String> wangyiCF = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油", "https://money.163.com/code/");
        });

        // 用anyOf合并为一个新的CompletableFuture
        CompletableFuture<Object> cfQuery = CompletableFuture.anyOf(sinaCF, wangyiCF);
        cfQuery.thenAccept((result) -> {
            System.out.println("price: " + result);
        });

        // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
        Thread.sleep(2000);
    }

    static String queryCode(String name, String url) {
        System.out.println("query code from " + url + "...");
        try {
            Thread.sleep((long) (Math.random() * 100));
        } catch (InterruptedException e) {
        }
        return "601857";
    }

}
